/* Payment Platform from Lucky Byte, Inc.
 *
 * Copyright (c) 2016 Lucky Byte, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.lucky_byte.pay.tpdu;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * 处理客户端请求
 */
public class Worker implements Runnable
{
	private static final Logger logger = LogManager.getLogger();

	private Properties properties;
	private int so_timeout;
	private Socket accept_socket;
	private Socket proxy_socket = null;

	public Worker(Socket accept_socket, Properties properties) {
		this.accept_socket = accept_socket;
		this.properties = properties;
		try {
			String timeout = properties.getProperty("tpdu.so_timeout");
			if (timeout != null) {
				this.so_timeout = Integer.parseInt(timeout);
			} else {
				this.so_timeout = 30;
			}
		} catch (NumberFormatException e) {
			logger.warn("配置项[tpdu.so_timeout]值[{}]无效，使用默认值 30.",
					properties.getProperty("tpdu.so_timeout"));
			this.so_timeout = 30;
		}
	}

	/**
	 * 从网络输入流读取报文前缀长度
	 */
	private short readPacketSize(DataInputStream data_is) throws IOException {
		byte[] bytes = new byte[2];
		data_is.readFully(bytes);

		ByteBuffer byte_buff = ByteBuffer.wrap(bytes);
		byte_buff.order(ByteOrder.BIG_ENDIAN);
		short pkt_size = byte_buff.getShort();
		if (pkt_size <= 22 || pkt_size > 4096) {
			logger.error("客户端[{}][{}]发送的报文长度[{}]有误.",
					accept_socket.getInetAddress().getHostAddress(),
					accept_socket.getPort(), pkt_size);
			return -1;
		}
		logger.info("读取报文前缀长度[{}].", pkt_size);
		return pkt_size;
	}

	/**
	 * 发送报文长度前缀到网络
	 */
	private void writePacketSize(DataOutputStream data_os, short pkt_size)
			throws IOException {
		ByteBuffer byte_buff = ByteBuffer.allocate(2);
		byte_buff.order(ByteOrder.BIG_ENDIAN);
		byte_buff.putShort(pkt_size);
		data_os.write(byte_buff.array());
		logger.info("发送报文前缀长度[{}].", pkt_size);
	}

	/**
	 * 从客户端读取请求报文
	 */
	private byte[] readReqtPacket(Socket socket) throws IOException {
		DataInputStream data_is =
				new DataInputStream(socket.getInputStream());
		short pkt_size = this.readPacketSize(data_is);
		if (pkt_size < 0) {
			return null;
		}
		byte[] bytes = new byte[pkt_size];
		data_is.readFully(bytes);
		logger.trace("接收客户端报文:\n{}", this.formatBytes(bytes));
		return bytes;
	}

	/**
	 * 获取请求报文中的 TPDU
	 */
	private String getTPDU(byte[] packet) {
		StringBuilder builder = new StringBuilder();
		if (packet.length < 5) {
			return null;
		}
		for (int i = 0; i < 5; i++) {
			builder.append(String.format("%02x", packet[i]));
		}
		return builder.toString();
	}

	/**
	 * 获取报文头中应用类别
	 */
	private String getAppType(byte[] packet) {
		return String.format("%02x", packet[5]);
	}

	/**
	 * 发送数据到后端处理，接收处理结果
	 */
	private byte[] sendRecv(byte[] packet, String host, int port) {
		try {
			if (proxy_socket == null) {
				proxy_socket = new Socket(host, port);
				proxy_socket.setSoTimeout(this.so_timeout * 1000);
				logger.info("与后端建立连接，本地地址[{}][{}]，后端地址[{}][{}].",
						proxy_socket.getLocalAddress().getHostAddress(),
						proxy_socket.getLocalPort(),
						proxy_socket.getInetAddress().getHostAddress(),
						proxy_socket.getPort());
			}
			if (!proxy_socket.isConnected()) {
				InetSocketAddress addr = new InetSocketAddress(host, port);
				proxy_socket.connect(addr, so_timeout);
			}
			DataOutputStream data_os = new DataOutputStream(
					proxy_socket.getOutputStream());
			this.writePacketSize(data_os, (short) packet.length);
			data_os.write(packet);
			data_os.flush();
			logger.trace("发送到后端报文:\n{}", this.formatBytes(packet));

			DataInputStream data_is = new DataInputStream(
					proxy_socket.getInputStream());
			short pkt_size = this.readPacketSize(data_is);
			if (pkt_size < 0) {
				return null;
			}
			byte[] bytes = new byte[pkt_size];
			data_is.readFully(bytes);
			logger.trace("从后端接收报文:\n{}", this.formatBytes(bytes));
			return bytes;
		} catch (EOFException e) {
			logger.error("与后端[{}][{}]通讯错误[后端关闭连接].", host, port);
			return null;
		} catch (Exception e) {
			logger.error("与后端[{}][{}]通讯错误[{}].", host, port,
					e.getMessage());
			return null;
		}
	}

	/**
	 * 将请求报文转发到后端服务处理
	 */
	private byte[] proxy(byte[] packet) {
		String tpdu = this.getTPDU(packet);
		if (tpdu == null) {
			logger.error("获取报文 TPDU 失败，丢弃报文.");
			return null;
		}
		String apptype = this.getAppType(packet);
		String[] keys = {
				tpdu + "." + apptype, "*." + apptype, tpdu + ".*"
		};
		String backend = null;
		for (String key : keys) {
			backend = properties.getProperty(key, null);
			if (backend != null) {
				logger.info("通过[{}]定位到后端服务[{}].", key, backend);
				break;
			}
		}
		if (backend == null) {
			logger.error("通过 TPDU[{}]+APPTYPE[{}]不能定位到后端服务.",
					tpdu, apptype);
			return null;
		}
		String[] strv = backend.split(";");
		if (strv.length < 2) {
			logger.error("TPDU[{}]对应的后端配置[{}]错误，请更正.",
					tpdu, backend);
			return null;
		}
		Map<String, String> map = new HashMap<>();
		for (int i = 0; i < strv.length; i++) {
			String[] kv = strv[i].split("=");
			if (kv.length != 2) {
				logger.error("TPDU[{}]对应的后端配置[{}]错误，请更正.",
						tpdu, backend);
				return null;
			}
			map.put(kv[0].trim(), kv[1].trim());
		}
		String host = map.get("host");
		String port = map.get("port");
		if (host == null || port == null) {
			logger.error("TPDU[{}]对应的后端配置[{}]缺少[host]或[port]项.",
					tpdu, backend);
			return null;
		}
		logger.info("映射[{}][{}]到后端[{}][{}].", tpdu, apptype, host, port);

		byte[] respkt = this.sendRecv(packet, host, Integer.parseInt(port));

		Map<String, Object> record = new HashMap<>();
		record.put("i_r_host", accept_socket.getInetAddress().getHostAddress());
		record.put("i_r_port", accept_socket.getPort());
		record.put("i_l_host", accept_socket.getLocalAddress().getHostAddress());
		record.put("i_l_port", accept_socket.getLocalPort());
		if (proxy_socket != null && proxy_socket.isConnected()) {
			record.put("o_r_host", proxy_socket.getInetAddress().getHostAddress());
			record.put("o_r_port", proxy_socket.getPort());
			record.put("o_l_host", proxy_socket.getLocalAddress().getHostAddress());
			record.put("o_l_port", proxy_socket.getLocalPort());
		}
		record.put("i_size", packet.length);
		if (respkt != null) {
			record.put("o_size", respkt.length);
		}
		Derby derby = Derby.getInstance();
		derby.insertRoutes(record);

		if (respkt == null) {
			if (proxy_socket != null) {
				try {
					proxy_socket.close();
				} catch (IOException e) {
					logger.catching(e);
				}
				proxy_socket = null;
			}
		}
		return respkt;
	}

    /**
     * 在单独的线程中执行的代码块
     */
	public void run() {
		while (true) {
			try {
				byte[] packet = this.readReqtPacket(accept_socket);
				if (packet == null) {
					logger.error("读取客户端[{}][{}]请求报文错误，放弃处理.",
							accept_socket.getInetAddress().getHostAddress(),
							accept_socket.getPort());
					accept_socket.close();
					break;
				}
				byte[] respkt = this.proxy(packet);
				if (respkt == null) {
					logger.error("处理客户端[{}][{}]的请求失败.",
							accept_socket.getInetAddress().getHostAddress(),
							accept_socket.getPort());
					accept_socket.close();
					break;
				}
				DataOutputStream data_os = new DataOutputStream(
						accept_socket.getOutputStream());
				this.writePacketSize(data_os, (short) respkt.length);
				data_os.write(respkt);
				data_os.flush();
				logger.trace("响应客户端报文:\n{}", this.formatBytes(respkt));
			} catch (EOFException e) {
				logger.debug("与客户端[{}][{}]通讯错误[客户端关闭连接].",
						accept_socket.getInetAddress().getHostAddress(),
						accept_socket.getPort());
				break;
			} catch (IOException e) {
				logger.error("与客户端[{}][{}]通讯错误[{}].",
						accept_socket.getInetAddress().getHostAddress(),
						accept_socket.getPort(), e.getMessage());
				break;
			}
		}
		logger.info("客户端[{}][{}]的请求处理完毕.",
				accept_socket.getInetAddress().getHostAddress(),
				accept_socket.getPort());
		try {
			if (proxy_socket != null) {
				proxy_socket.close();
				proxy_socket = null;
			}
			accept_socket.close();
		} catch (IOException e) {
			logger.catching(e);
		}
	}

	/**
	 * 格式化显示二进制数据
	 */
	public String formatBytes(byte[] bytes, int offset, int length) {
		StringBuilder builder = new StringBuilder("");
		if (length <= 0) {
			length = bytes.length;
		}
		for (int i = offset, j = 1; i < length; i++, j++) {
			builder.append(String.format("%02X", bytes[i]));
			if (j % 24 == 0) {
				builder.append("\n");
			} else if (j % 8 == 0) {
				builder.append("   ");
			} else if (i < length) {
				builder.append(" ");
			}
		}
		return builder.toString();
	}

	/**
	 * 上面函数的简化版本
	 */
	private String formatBytes(byte[] bytes) {
		return this.formatBytes(bytes, 0, -1);
	}
}
